package com.mq.mqserver.core;

import com.mq.common.Consumer;
import com.mq.common.ConsumerEnv;
import com.mq.common.MqException;
import com.mq.mqserver.VirtualHost;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * 实现 消费消息 的核心逻辑
 * 服务器每收到一条消息,就将其放到一个线程池中,消费者创建后,按照轮训的方式依次从线程池中取消息,进行消费
 */
public class ConsumerManager {
    //持有上层的VirtualHost对象的引用,用来操作数据
    private VirtualHost virtualHost;
//    使⽤⼀个线程池⽤来执⾏消息回调
    private ExecutorService workerPool = Executors.newFixedThreadPool(4);
    //存放令牌(队列名)的队列:那个队列当前有消息了,就将队列名加入到阻塞队列中
    //然后扫描线程通过该队列中存放的队列名找到对应的消息和订阅者,将信息打包放到线程池中进行消费
    private BlockingQueue<String> tokenQueue = new LinkedBlockingQueue<>();
    //扫描线程
    private Thread scannerThread = null;


    //先获取到令牌,根据令牌找到指定的队列,从队列中获取消息进行消费
    public ConsumerManager(VirtualHost parent){
        virtualHost = parent;
        //为推的模式.不断的扫描令牌队列,一但有消息进入队列,就将其推送给消费者
        Thread t = new Thread(()->
        {
            while(true){
                try {
                    //1.获取令牌
                    String queueName = tokenQueue.take();
                    //2.根据令牌,找到指定的队列
                    MSGQueue queue = virtualHost.getMemoryDataCenter().getQueue(queueName);
                    if(queue==null){
                        throw new MqException("[ConsumerManager] 获取令牌时,发现队列不存在");
                    }
                    synchronized (queue){
                    //3.从队列中获取一个消息并进行消费
                        consumerMessage(queue);
                    }
                } catch (InterruptedException | MqException e) {
                    throw new RuntimeException(e);
                }
            }
        });
        //将线程设为后台线程
        //当前台线程执行结束了,后台线程也就结束了,
        //若设为前台线程,那么只有当前台线程执行完了,整个进程才会结束,
        // 这里的循环是while(true)会一直卡着执行结束不了,因此要设成后台线程
        t.setDaemon(true);
        //启动线程
        t.start();
    }

    //通知消费者消费消息
    //调用时机:发送方发送消息成功后,
    //当队列中有消息了,就将其放到阻塞队列中,然后就要通知消费者消费消息了
    public void notifyConsumer(String queueName) throws InterruptedException {
        tokenQueue.put(queueName);
    }

    //  消费消息:
    //从队列中获取一个消息,并让消费者消费,
    // 当消费者不止一个时,按照轮训的方式让消费者依次消费消息
    private void consumerMessage(MSGQueue queue) throws MqException {
        //1.从队列的订阅者中挑选一个订阅者
        ConsumerEnv consumerEnv = queue.chooseConsumerEnv();
        if(consumerEnv==null){
            //当前队列号没有订阅者,无法消费消息
            System.out.println("[ConsumerManager] 当前队列中还没有订阅者");
            return;
        }
        //2.从队列中挑选一个消息
        Message message = virtualHost.getMemoryDataCenter().pollMessageFromQueue(queue.getName());
        if(message==null){
            //当前队列中还没有消息,不需要消费
            System.out.println("当前队列中还没有消息");
            return;
        }
        //将消息带到消费者的回调方法中,对给线程池执行
        workerPool.submit(()->{
            try{
                //1.在执行回调之前,先将消息放到待确认队列集合中,一旦消息被消费失败了.就重新发送消息
                virtualHost.getMemoryDataCenter().insertQueueMessageWaitAck(queue.getName(),message);
                //2.执行订阅者的回调方法
                consumerEnv.getConsumer().handlerDeliver(consumerEnv.getConsumerTag(),message.getBasicProperties(),message.getBody());
                //3.根据消费者的确认消息方式及消费者消费消息的情况,执行删除消息操作
                //  这里完成为自动确认模式下的操作,手动模式下,在basicAck方法中实现
                if(consumerEnv.isAutoAck()){
                    //4.删除磁盘中的数据
                    //  是否持久化
                    //  1:非持久化 0:持久化
                    if(message.getDeliveryMode()==0) {
                        virtualHost.getDiskDataCenter().deleteMessageFromQueue(queue, message);
                    }
                    //5.删除未确认消息队列中的消息
                    virtualHost.getMemoryDataCenter().deleteWaitAckQueue(queue.getName(), message.getMessageId());
                    //6.删除消息集合中的消息
                    virtualHost.getMemoryDataCenter().deleteMessage(message.getMessageId());
                    System.out.println("[ConsumerManager] 消息被成功消费 ");
                }
            }catch (Exception e){
                System.out.println("[ConsumerManager] 消费消息失败");
                e.printStackTrace();
            }
        });
    }

    //向要订阅的队列添加消费者对象,
    // 并且当队列中已经有消息时,要将其消费了
    public void addConsumer(String consumerTag, String queueName, boolean autoAck, Consumer consumer) throws MqException {
        //1.找到对应的队列
        MSGQueue queue = virtualHost.getMemoryDataCenter().getQueue(queueName);
        if(queue==null){
            throw new MqException("[ConsumerManager] 队列不存在 queueName:"+queueName);
        }
        //2.创建一个消费者
        ConsumerEnv consumerEnv = new ConsumerEnv();
        consumerEnv.setConsumerTag(consumerTag);
        consumerEnv.setQueueName(queueName);
        consumerEnv.setAutoAck(autoAck);
        consumerEnv.setConsumer(consumer);
        //3.将订阅者加入到队列的订阅者队列中
        queue.addConsumerEnv(consumerEnv);
        //4.当队列中已经有一些消息时,要将其消费掉
        synchronized (queue){
            int n = virtualHost.getMemoryDataCenter().getMessageCountFromQueue(queueName);
            for(int i=0;i<n;i++){
                //这个方法调用一次就消费一条消息
                consumerMessage(queue);
            }
        }
    }
}
